package app_kvEcs;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Random;

import org.apache.log4j.Logger;

import client.CommunicationLogic;
import common.messages.KVMessage.StatusType;
import common.messages.MetaData;

public class ECS {
	private Logger logger = Logger.getRootLogger();

	public MetaData metaData = null;
	private ArrayList<ServerInfo> serverList = new ArrayList<ServerInfo>();

	private BufferedReader stdBF = new BufferedReader(new InputStreamReader(
			System.in));

	private String JAR_DIR = "/Users/uniquehuang/Documents/MS3_set/ms3-server.jar";

	ArrayList<ServerInfo> result;

	public ECS() {
		// Read configuration file and store the ip and port information
		File configurationFile = new File("ecs.config");
		BufferedReader bf;
		try {
			bf = new BufferedReader(new FileReader(configurationFile));
			String line = null;
			// Read server infos
			while ((line = bf.readLine()) != null) {
				// Split
				String split[] = line.split(" ");
				if (split.length != 3) {
					System.out
							.println("ERROR! Illegel configuration file format");
					System.exit(0);
				}
				String serverName = split[0];
				String serverAdd = split[1];
				String serverPort = split[2];

				// Save each server info into a list
				ServerInfo serverInfo = new ServerInfo(serverName, serverAdd,
						serverPort);
				serverList.add(serverInfo);
			}
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	/**
	 * Constructor
	 * 
	 * @param string
	 *            filename of configuration file
	 */
	public ECS(String string) {
		// Read configuration file and store the ip and port information
		File configurationFile = new File(string);
		try {
			BufferedReader bf = new BufferedReader(new FileReader(
					configurationFile));
			String line = null;
			// Read server infos
			while ((line = bf.readLine()) != null) {
				// Split
				String split[] = line.split(" ");
				if (split.length != 3) {
					System.out
							.println("ERROR! Illegel configuration file format");
					System.exit(0);
				}
				String serverName = split[0];
				String serverAdd = split[1];
				String serverPort = split[2];

				// Save each server info into a list
				ServerInfo serverInfo = new ServerInfo(serverName, serverAdd,
						serverPort);
				serverList.add(serverInfo);
			}

			// Command line user interface
			CommandLine();
		} catch (FileNotFoundException e) {
			System.out.println("ERROR! Cannot read configuration file");
			logger.error("ERROR! Cannot read configuration file");
			return;
		} catch (IOException e) {
			System.out.println("ERROR! IO");
			logger.error("ERROR! IO");
			return;
		}
	}

	/**
	 * User interface
	 */
	private void CommandLine() {
		while (true) {
			System.out.print("ECS>> ");
			String stdStr = readSTD();
			String token[] = stdStr.split(" ");

			if (token[0].equals("initService")) {
				// Validate the command
				if (token.length != 2) {
					System.out.println("Illegal number of parameters");
					continue;
				}
				int m = 0;
				try {
					m = Integer.parseInt(token[1]);
				} catch (Exception e) {
					System.out.println("Not a number");
					continue;
				}

				initService(m);
			} else if (token[0].equals("start")) {
				// Start the selected servers
				start();
			} else if (token[0].equals("stop")) {
				stop();
			} else if (token[0].equals("shutDown")) {
				shutDown();
			} else if (token[0].equals("addNode")) {
				addNode();
			} else if (token[0].equals("removeNode")) {
				removeNode();
			} else if (token[0].equals("quit")) {
				break;
			}
		}
		System.out.println("ECS is stopped.");
	}

	/**
	 * Read from standard input
	 * 
	 * @return
	 */
	private String readSTD() {
		String stdStr = "";
		try {
			stdStr = stdBF.readLine();
		} catch (IOException e) {
			System.out.println("ERROR! Failed to read from standard input");
			logger.error("ERROR! Failed to read from standard input");
			return null;
		}
		return stdStr;
	}

	/**
	 * Randomly choose <numberOfNodes> servers from the available machines and
	 * start the KVServer by issuing a SSH call to the respective machine. This
	 * call launches the server.
	 * 
	 * @param numberOfNodes
	 */
	public void initService(int numberOfNodes) {
		// Number of nodes cannot be bigger than the max number
		if (numberOfNodes > serverList.size()) {
			System.out.println("Don't have enough servers");
			return;
		}
		// Random numberOfNodes of server to run
		Random random = new Random();
		// Clone a copy of server list
		ArrayList<ServerInfo> temp = new ArrayList<ServerInfo>();
		for (ServerInfo info : serverList) {
			temp.add(info.clone());
		}
		result = new ArrayList<ServerInfo>();
		int size = temp.size();
		for (int i = 0; i < numberOfNodes; i++) {
			// Get a random position
			int position = random.nextInt(size - i);
			// Store
			result.add(temp.get(position).clone());
			// Replace the position with the last unused element, if it is the
			// last one, don't do it
			if (i != numberOfNodes - 1) {
				temp.remove(position);
				temp.add(position, temp.get(size - 2 - i).clone());
			}
		}
		// Generate consistent hash ring and corresponding meta data
		metaData = new MetaData();
		// Run script to open server program in the remote computer
		Process proc;
		// TODO ssh connection without password
		for (ServerInfo info : result) {
			String command = "";
			command += "java -jar ";
			command += JAR_DIR + " ";
			command += info.getPort() + "  ALL\n";

			Runtime run = Runtime.getRuntime();
			try {
				System.out.println(command);
				proc = run.exec(command);
			} catch (IOException e) {
				System.out.println("ERROR! Failed to run script");
				logger.error("ERROR! Failed to run script");
				return;
			}
			// Add one meta data
			metaData.add(info.getAdd() + ":" + info.getPort());
		}
		// Wait for latency
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		// Communicate to each server and send meta data to each of them
		for (ServerInfo info : result) {
			CommunicationLogic communicate = new CommunicationLogic(
					info.getAdd(), Integer.parseInt(info.getPort()));
			try {
				communicate.connect();
				String rec_msg = communicate.receive();
				System.out.println(rec_msg);
				communicate.send("ECS initService \n" + metaData.toString());
				String ackMsg = communicate.receive();
				if (ackMsg.equals("ack")) {
					System.out.println("ECS>> Meta data sended to server "
							+ info.getAdd() + ":" + info.getPort()
							+ " successfully");
				} else {
					System.out
							.println("ERROR! Failed to send meta data to server "
									+ info.getAdd() + ":" + info.getPort());
				}
				communicate.disconnect();
			} catch (IOException e) {
				System.out
						.println("ERROR! Cannot connect to server, operation aborted");
				logger.error("ERROR! Cannot connect to server, operation aborted");
				return;
			}

		}
	}

	/**
	 * Starts the storage service by calling start() on all KVServer instances
	 * that participate in the service.
	 */
	public void start() {
		// Send start message to every server
		for (Iterator<String> it = metaData.getCircle().keySet().iterator(); it
				.hasNext();) {
			String k = it.next();
			String value = metaData.getCircle().get(k);
			String split[] = value.split(":");
			// Connect to server
			CommunicationLogic communication = new CommunicationLogic(split[0],
					Integer.parseInt(split[1]));
			try {
				communication.connect();
				String rec_msg = communication.receive();
				System.out.println(rec_msg);
				// Send start command to server
				communication.send("ECS start");
				String ackMsg = communication.receive();
				// Receive ack
				if (ackMsg.equals("ack")) {
					System.out.println("Srart server successfully");
				}
				communication.disconnect();
			} catch (IOException e) {
				System.out.println("ERROR! Cannot connect to server");
				logger.error("ERROR! Cannot connect to server");
			}
		}
	}

	/**
	 * Stops the service; all participating KVServers are stopped for processing
	 * client requests but the processes remain running.
	 */
	public void stop() {
		// Send stop message to every server
		for (Iterator<String> it = metaData.getCircle().keySet().iterator(); it
				.hasNext();) {
			String k = it.next();
			String value = metaData.getCircle().get(k);
			String split[] = value.split(":");
			// Connect to server
			CommunicationLogic communication = new CommunicationLogic(split[0],
					Integer.parseInt(split[1]));
			try {
				communication.connect();
				String rec_msg = communication.receive();
				System.out.println(rec_msg);
				// Send stop command to server
				communication.send("ECS stop");
				String ackMsg = communication.receive();
				// Receive ack
				if (ackMsg.equals("ack")) {
					System.out.println("Stop server successfully");
				}
				communication.disconnect();
			} catch (IOException e) {
				System.out.println("ERROR! Cannot connect to server");
				logger.error("ERROR! Cannot connect to server");
			}
		}
	}

	/**
	 * Stops all server instances and exits the remote processes.
	 */
	public void shutDown() {
		// Send shutdown message to every server
		for (Iterator<String> it = metaData.getCircle().keySet().iterator(); it
				.hasNext();) {
			String k = it.next();
			String value = metaData.getCircle().get(k);
			String split[] = value.split(":");
			// Connect to server
			CommunicationLogic communication = new CommunicationLogic(split[0],
					Integer.parseInt(split[1]));
			try {
				communication.connect();
				String rec_msg = communication.receive();
				System.out.println(rec_msg);
				// Send shutdown command to server
				communication.send("ECS shutDown");
				String ackMsg = communication.receive();
				// Receive ack
				if (ackMsg.equals("ack")) {
					System.out.println("Shutdown server successfully");
				}
				communication.disconnect();
			} catch (IOException e) {
				System.out.println("ERROR! Cannot connect to server");
				logger.error("ERROR! Cannot connect to server");
			}
		}
		// Remove server from meta data according to unavailable server
		// list(result)
		for (ServerInfo info : result) {
			metaData.remove(info.getAdd() + ":" + info.getPort());
		}
		// Remove servers from unavailable list(result)
		result.removeAll(result);
	}

	/**
	 * Add a new node to the storage service at an arbitrary position.
	 */
	public void addNode() {
		// Clone a copy of server list
		ArrayList<ServerInfo> temp = new ArrayList<ServerInfo>(serverList);
		// Leave out unavailable servers
		Iterator<ServerInfo> it = temp.iterator();
		while (it.hasNext()) {
			ServerInfo info = it.next();
			Iterator<ServerInfo> itResult = result.iterator();
			while (itResult.hasNext()) {
				ServerInfo infoResult = itResult.next();
				if (info.equals(infoResult)) {
					it.remove();
				}
			}
		}
		// Random numberOfNodes of server to run
		Random random = new Random();
		// Random number according to size
		int seed = temp.size();
		// No more server
		if (seed == 0) {
			System.out.println("No more available server");
			return;
		}
		int randomIndex = random.nextInt(seed);
		// Start a server
		// TODO should use SSH
		String command = "";
		command += "java -jar ";
		command += JAR_DIR + " ";
		command += temp.get(randomIndex).getPort() + "  ALL\n";
		// Run script to open server program in the remote computer
		Process proc;
		Runtime run = Runtime.getRuntime();
		try {
			System.out.println(command);
			proc = run.exec(command);
		} catch (IOException e) {
			System.out.println("ERROR! Failed to run script");
			logger.error("ERROR! Failed to run script");
			return;
		}
		// Find the next server in the consistent hash ring and store it
		String nextServer = metaData.get(temp.get(randomIndex).getAdd() + ":"
				+ temp.get(randomIndex).getPort());
		// Add one meta data
		metaData.add(temp.get(randomIndex).getAdd() + ":"
				+ temp.get(randomIndex).getPort());
		// Wait for latency
		try {
			Thread.sleep(3000);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		// Communicate server
		CommunicationLogic communicate = new CommunicationLogic(temp.get(
				randomIndex).getAdd(), Integer.parseInt(temp.get(randomIndex)
				.getPort()));
		try {
			communicate.connect();
			String rec_msg = communicate.receive();
			System.out.println(rec_msg);
			communicate.send("ECS addNode \n" + metaData.toString());
			String ackMsg = communicate.receive();
			if (ackMsg.equals("ack")) {
				System.out.println("ECS>> Meta data sended to server "
						+ temp.get(randomIndex).getAdd() + ":"
						+ temp.get(randomIndex).getPort() + " successfully");
			} else {
				System.out.println("ERROR! Failed to send meta data to server "
						+ temp.get(randomIndex).getAdd() + ":"
						+ temp.get(randomIndex).getPort());
			}
			communicate.disconnect();
		} catch (IOException e) {
			System.out
					.println("ERROR! Cannot connect to server, operation aborted");
			logger.error("ERROR! Cannot connect to server, operation aborted");
			return;
		}
		// Update meta data to all server
		for (ServerInfo info : result) {
			// Communicate server
			CommunicationLogic communicateUpdate = new CommunicationLogic(
					info.getAdd(), Integer.parseInt(info.getPort()));
			try {
				communicateUpdate.connect();
				String rec_msg = communicateUpdate.receive();
				System.out.println(rec_msg);
				communicateUpdate.send("ECS updateMetaData \n"
						+ metaData.toString());
				String ackMsg = communicateUpdate.receive();
				if (ackMsg.equals("ack")) {
					System.out.println("ECS>> Meta data update to server "
							+ info.getAdd() + ":" + info.getPort()
							+ " successfully");
				} else {
					System.out
							.println("ERROR! Failed to send meta data to server "
									+ info.getAdd() + ":" + info.getPort());
				}
				communicateUpdate.disconnect();
			} catch (IOException e) {
				System.out
						.println("ERROR! Cannot connect to server, operation aborted");
				logger.error("ERROR! Cannot connect to server, operation aborted");
				return;
			}
		}
		// Add new node to unavailable list(result)
		ServerInfo randomServer = temp.get(randomIndex);
		result.add(new ServerInfo(randomServer.getName(),
				randomServer.getAdd(), randomServer.getPort()));
		// Move data from next server
		String splitNextServer[] = nextServer.split(":");
		// Communicate next server
		CommunicationLogic communicateNext = new CommunicationLogic(
				splitNextServer[0], Integer.parseInt(splitNextServer[1]));
		try {
			communicateNext.connect();
			String rec_msg = communicateNext.receive();
			System.out.println(rec_msg);
			communicateNext.send("ECS moveDataTo "
					+ temp.get(randomIndex).getAdd() + ":"
					+ temp.get(randomIndex).getPort());
			String ackMsg = communicateNext.receive();
			if (ackMsg.equals("ack")) {
				System.out.println("ECS>> Data already moved to server "
						+ splitNextServer[0] + ":" + splitNextServer[1]
						+ " successfully");
			} else {
				System.out.println("ERROR! Failed to send meta data to server "
						+ splitNextServer[0] + ":" + splitNextServer[1]);
			}
			communicateNext.disconnect();
		} catch (IOException e) {
			System.out
					.println("ERROR! Cannot connect to server, operation aborted");
			logger.error("ERROR! Cannot connect to server, operation aborted");
			return;
		}
	}

	/**
	 * Remove a node from the storage service at an arbitrary position.
	 */
	public void removeNode() {
		// Random a server to remove from unavailable server list(result)
		int seed = result.size();
		Random random = new Random();
		int randomPosition = random.nextInt(seed);
		ServerInfo randomServer = result.get(randomPosition);
		String randomServerIP = randomServer.getAdd();
		String randomServerPort = randomServer.getPort();
		// Remove node from meta data
		metaData.remove(randomServerIP + ":" + randomServerPort);
		// Update meta data to all server
		for (ServerInfo info : result) {
			// Communicate server
			CommunicationLogic communicateUpdate = new CommunicationLogic(
					info.getAdd(), Integer.parseInt(info.getPort()));
			try {
				communicateUpdate.connect();
				String rec_msg = communicateUpdate.receive();
				System.out.println(rec_msg);
				communicateUpdate.send("ECS updateMetaData \n"
						+ metaData.toString());
				String ackMsg = communicateUpdate.receive();
				if (ackMsg.equals("ack")) {
					System.out.println("ECS>> Meta data update to server "
							+ info.getAdd() + ":" + info.getPort()
							+ " successfully");
				} else {
					System.out
							.println("ERROR! Failed to send meta data to server "
									+ info.getAdd() + ":" + info.getPort());
				}
				communicateUpdate.disconnect();
			} catch (IOException e) {
				System.out
						.println("ERROR! Cannot connect to server, operation aborted");
				logger.error("ERROR! Cannot connect to server, operation aborted");
				return;
			}
		}
		// Get next server of the random server
		String nextServer = metaData.get(randomServerIP + ":"
				+ randomServerPort);
		String split[] = nextServer.split(":");
		String nextServerIP = split[0];
		String nextServerPort = split[1];
		// Communicate to the random server and send the next server's address
		// to it, then it will remove all the data to the next server itself
		CommunicationLogic communicateRandomServer = new CommunicationLogic(
				randomServerIP, Integer.parseInt(randomServerPort));
		try {
			communicateRandomServer.connect();
			String rec_msg = communicateRandomServer.receive();
			System.out.println(rec_msg);
			communicateRandomServer.send("ECS removeNode to " + nextServerIP
					+ ":" + nextServerPort);
			String ackMsg = communicateRandomServer.receive();
			if (ackMsg.equals("ack")) {
				System.out.println("ECS>> Server" + randomServerIP + ":"
						+ randomServerPort + " removed successfully ");
			} else {
				System.out.println("ERROR! Failed to remove server "
						+ randomServerIP + ":" + randomServerPort);
			}
			communicateRandomServer.disconnect();
		} catch (IOException e) {
			System.out
					.println("ERROR! Cannot connect to server, operation aborted");
			logger.error("ERROR! Cannot connect to server, operation aborted");
			return;
		}
		// Remove server from unavailable server list(result)
		result.remove(randomPosition);

	}
}
